package data.basic

import data.bean.Domain.Access
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

// 开发时要做的第一件事情
import org.apache.flink.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api._

import org.apache.flink.table.api.Table
import org.apache.flink.types.Row

object DataStreamTableSQLApp {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnvironment = StreamTableEnvironment.create(env)

    val stream = env.readTextFile("data/access.log")
      .map(x => {
        val splits = x.split(",")
        Access(splits(0).trim.toLong, splits(1).trim, splits(2).trim.toDouble)
      })

    // datastream ==> table
    val accessTable: Table = tableEnvironment.fromDataStream(stream)
    tableEnvironment.createTemporaryView("access", accessTable)

    val resultTable: Table = tableEnvironment.sqlQuery("select * from access where domain='bsdata.com'")

      // table ==> datastream  3种方式
      tableEnvironment.toAppendStream[Row](resultTable).print("==1==")
      tableEnvironment.toAppendStream[(Long,String,Double)](resultTable).print("==2==")  // 方式2
      resultTable.toAppendStream[(Long,String,Double)].print("==3==")  // 方式3


    accessTable.select("*").toAppendStream[Row].print("==所有字段==")
    accessTable.select("domain").toAppendStream[Row].print("==单个字段过时写法==")
    accessTable.select('domain).toAppendStream[Row].print("==单个字段推荐写法==")  // 推荐

    accessTable.select("time,domain,traffic").toAppendStream[Row].print("==多个字段过时写法==")
    accessTable.select('time,'domain,'traffic).toAppendStream[Row].print("==多个字段推荐写法==")  // 推荐

      // fromDataStream 方式2
    val accessTable2: Table = tableEnvironment.fromDataStream(stream, 'a, 'b, 'c)
    accessTable2.printSchema()
    accessTable2.select('a, 'b, 'c).toAppendStream[Row].print()

    env.execute(getClass.getCanonicalName)
  }
}
